package com.stress.testing.plugin.apache.rocketmq.interceptor;

import com.stress.testing.common.StressTesting;
import com.stress.testing.common.interceptor.AbstractInterceptor;
import com.stress.testing.plugin.apache.rocketmq.context.PushConsumerContext;
import com.stress.testing.plugin.apache.rocketmq.context.PushConsumerContextHolder;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;

import java.lang.reflect.Method;
import java.util.Arrays;

import static com.stress.testing.common.conf.PluginConfigKey.DEFAULT_PLUGIN_APACHE_ROCKETMQ_TOPIC_PREFIX;
import static com.stress.testing.common.conf.PluginConfigKey.PLUGIN_APACHE_ROCKETMQ_TOPIC_PREFIX_KEY;

public class DefaultMQPushConsumerSubscribeInterceptor extends AbstractInterceptor {
    @Override
    public void before(Method method, Object[] args, Object zuper, Object thiz) {
        if (!StressTesting.enable(getAgentConfig(), getPluginConfig())) {
            return;
        }

        if (args == null || args.length == 0) {
            return;
        }

        Object arg = args[0];

        if (arg == null || !(arg instanceof String)) {
            return;
        }

        if (thiz == null || !(thiz instanceof DefaultMQPushConsumer)) {
            return;
        }

        DefaultMQPushConsumer consumer = (DefaultMQPushConsumer) thiz;

        String topicPrefix = getPluginConfig().get(PLUGIN_APACHE_ROCKETMQ_TOPIC_PREFIX_KEY,
                DEFAULT_PLUGIN_APACHE_ROCKETMQ_TOPIC_PREFIX);
        String topicOrigin = (String) arg;
        if (topicOrigin != null && topicOrigin.startsWith(topicPrefix)) {
            return;
        }

        Object[] newArgs = Arrays.copyOf(args, args.length);
        String newTopic = topicPrefix + topicOrigin;
        newArgs[0] = newTopic;

        PushConsumerContext context = PushConsumerContextHolder.getPushConsumerContext(consumer.hashCode());
        if (context == null) {
            context = new PushConsumerContext();
            PushConsumerContextHolder.setPushConsumerContext(consumer.hashCode(), context);
        }

        context.setSubscribeArgs(newArgs);
    }
}
